pyannote.audioを使って誰がいつ話したのかを判定する話者ダイアライゼーションをやってみた

これで誰かの声だけ見つけちゃおう

#Python

#機械学習

#Deep Learning

nokomoro3

2023.04.06

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんちには。

データアナリティクス事業本部インテグレーション部機械学習チームの中村です。

今回はpyannote.audioで、誰がいつ話したのかを判定する話者ダイアライゼーションをやってみたいと思います。

話者ダイアライゼーションとは

話者ダイアライゼーションとは、どこの時間でどの話者がしゃべったのか、話者認識をせずに実施する技術のことを指します。

話者認識（Speaker Recognition）は、音声から個人までを特定する話者識別（Speaker Identification）や話者検証（Speaker Verification）から構成されますが、話者ダイアライゼーション（Speaker Diarization）は個人を特定はせず、発話者を区別するのみとなります。

またどこからどこまでで発話したのかの時間情報を出力するのも話者ダイアライゼーションの特徴です。

ちなみに話者分離（Speaker Seperation）という言葉もありますが、こちらは音響的に信号を分離する話を指しています。つまり混ざった音声を入力すると、それぞれの話者の音声だけ抽出されたデータを得るタスクになります。

説明が長くなりましたが、今回は話者ダイアライゼーションに関する内容です。

pyannote.audioとは

pyannote.audioとは前述の話者ダイアライゼーションを行うライブラリです。

GitHubは以下となります。

ベースのフレームワークはPyTorchとなっており、end-to-endの話者ダイアライゼーションを実現します。

使用環境

Google Colaboratoryを使います。ハードウェアアクセラレータはGPU、GPUのタイプはT4、ラインタイム仕様は標準で実施します。

主なバージョン情報は以下です。

!python --version

Python 3.9.16

パッケージのバージョンは以下です。（こちらは準備を一通り実施した後に確認できます）

!pip freeze | grep \
    -e "pydub" -e "polars" \
    -e "torch==" -e "torchvision" -e "torchaudio" -e "torchtext" \
    -e "speechbrain" -e "pyannote.audio" -e "ipython=="

ipython==7.34.0
polars==0.16.18
pyannote.audio==2.1.1
pydub==0.25.1
speechbrain==0.5.12
torch==1.11.0
torchaudio==0.11.0
torchtext==0.12.0
torchvision==0.12.0

準備

Hugging Faceコンソールでの操作

まずはHugging Faceのアカウントを作成してログインします。

その後、以下２つのリポジトリのuser conditionをAcceptする必要があります。

Accept画面は以下のように表示されるので、必要事項を入力してボタンを押下してください。

その後、ユーザ管理画面の以下のURLにアクセスし、トークンを作成しておきます。

https://huggingface.co/settings/tokens

環境設定

先にpyannote.audio意外のモジュールをインストールします。（pyannote.audioインストール後は!で始まるコマンドがエラーとなるため、原因は不明。）

!pip install pydub
!pip install polars

その後、pyannote.audioに関するインストールをします。

# for speechbrain
!pip install -qq torch==1.11.0 torchvision==0.12.0 torchaudio==0.11.0 torchtext==0.12.0
!pip install -qq speechbrain==0.5.12

# pyannote.audio
!pip install -qq pyannote.audio

# for visualization purposes
!pip install -qq ipython==7.34.0

インストールは、以下の公式GitHubにあるColaboratoryのノートブックを参考にしています。

その後、モジュールをインポートします。

from pyannote.audio import Pipeline
from pyannote.core import Segment, notebook, Annotation
import json
from pydub import AudioSegment
import polars as pl

モデル取得

以下でHugging Faceからモデルを取得します。

pipeline = Pipeline.from_pretrained(
    "pyannote/[email protected]"
    , use_auth_token="<Hugging Faceコンソールでの操作で取得したトークンを記載>"
)

その際、トークンはここまでで作成済みのものを使用してください。

話者判別の実施

準備が整ったので、判別処理をしてみます。

今回はクラスメソッド公式の以下の動画を使用しました。

こちらの音声データをaudio.wavとして保存してあるとして以降を実行します。

%%time
diarization = pipeline("audio.wav")

CPU times: user 3min 19s, sys: 3.01 s, total: 3min 22s
Wall time: 3min 26s

動画は約1時間のデータですが、約3分30秒で処理が完了しました。

結果は以下のようにAnnotationクラスとして取得されます。

type(diarization)

pyannote.core.annotation.Annotation

補足：話者数の設定した分析

話者数があらかじめわかっている場合は、pipelineに設定することでより想定通りに判別をさせることが可能です。

今回は実行しませんが、コードのみ記載します。

diarization = pipeline("audio.wav", num_speakers=2)

minとmaxという形で指定することもできます。こちらもコードのみ記載します。

diarization = pipeline("audio.wav", min_speakers=2, max_speakers=5)

グラフ表示

notebookでは以下で全体の結果を確認できます。

diarization

この表示範囲を狭めたい場合は、以下のように制限を掛けることができます。

start = 200
EXCERPT = Segment(start, start+60)
notebook.crop = EXCERPT
diarization

表示範囲を元に戻す場合は、resetが必要です。

notebook.reset()

結果の確認

各セグメントの結果を見るためには、以下のようにgeneratorとしてループさせることで確認ができます。

for segment, track_name, label in diarization.itertracks(yield_label=True):
    print(f"{segment.start=:.1f}, {segment.end=:.1f}, {track_name=}, {label=}")

segment.start=0.5, segment.end=0.6, track_name='FS', label='SPEAKER_02'
segment.start=3.4, segment.end=7.0, track_name='FT', label='SPEAKER_02'
segment.start=5.5, segment.end=6.1, track_name='A', label='SPEAKER_00'
segment.start=7.9, segment.end=8.8, track_name='FU', label='SPEAKER_02'
segment.start=9.6, segment.end=46.3, track_name='FV', label='SPEAKER_02'
…(以降略)…

このうちstart - stopが発話の区間を示し、labelがその区間の話者を示します。

track_nameは発話セグメントのIDのようなもののようで、今回は特に使用しません。

統計

各話者の発話の総時間などはchartで確認できます。

sorted(diarization.chart())

[('SPEAKER_00', 79.63312500000063),
 ('SPEAKER_01', 263.01374999999786),
 ('SPEAKER_02', 1039.1118750000023),
 ('SPEAKER_03', 829.743750000001),
 ('SPEAKER_04', 294.7387499999977),
 ('SPEAKER_05', 913.9162500000035),
 ('SPEAKER_06', 46.929374999999546)]

ラベルの付け替え

今はSPEAKER_数字という形にラベル付けされていますが、以下のようにすれば結果に対してラベルを後編集することも可能になっています。

diarization_renamed = diarization.rename_labels({"SPEAKER_06": "noise"})
diarization_renamed.chart()

[('SPEAKER_02', 1039.1118750000023),
 ('SPEAKER_05', 913.9162500000035),
 ('SPEAKER_03', 829.743750000001),
 ('noise', 294.7387499999977),
 ('SPEAKER_01', 263.01374999999786),
 ('SPEAKER_00', 79.63312500000063),
 ('SPEAKER_06', 46.929374999999546)]

結果の永続化

結果はfor_jsonという関数でdict型で取得することも可能です。

ですのでdict型にした後、jsonモジュールでファイルに出力して永続化することが可能です。

with open("diarization.json", "wt") as f:
    json.dump(diarization.for_json(), f)

保存した結果を、再度読み込みたい場合は以下のようにします。

with open("diarization.json", "rt") as f:
    diarization_reload = Annotation.from_json(json.load(f))

Pydubによる編集

最後に評価のためにPydubを使って同一話者の発話をそれぞれのファイルとして合成します。

セグメントの結果自体は、辞書型のcontentに入っているため、まずはpolarsを使ってDataFrameにします。

df = pl.DataFrame(diarization.for_json()["content"])
df

こちらを元に、Pydubによる切り出しと連結を以下のように行います。

audio_segment: AudioSegment = AudioSegment.from_file("audio.wav", format="wav")

for label in sorted(df["label"].unique()):

    # ある話者のデータを抽出
    _df = df.filter(pl.col("label") == label)

    # 初期化
    audio_segment_each_speaker = audio_segment[0:0]

    # セグメントのループ
    for i, segment in enumerate(_df["segment"]):

        # pyannote.audioの結果は秒、Pydubはミリ秒で範囲を指定するので、1000倍する
        start = segment["start"]*1000
        end = segment["end"]*1000

        # +=で連結することができる（便利！）
        audio_segment_each_speaker += audio_segment[start:end]

    print(f"{label=:}, duration={audio_segment_each_speaker.duration_seconds}")

    # exportで音声ファイルを出力
    audio_segment_each_speaker.export(f"{label=:}.wav", format="wav")

コメントにある程度説明を記載しましたが、Pydubの詳しい使い方についてはそのGitHubのページを参照ください。

以下が結果として得られるそれぞれの話者の音声データです。